package com.zhang.edu.realtime.app.dim;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.ververica.cdc.connectors.mysql.source.MySqlSource;
import com.ververica.cdc.connectors.mysql.table.StartupOptions;
import com.ververica.cdc.debezium.JsonDebeziumDeserializationSchema;
import com.zhang.edu.realtime.app.func.DimBroadcastFunction;
import com.zhang.edu.realtime.app.func.DimPhoenixSinkFunc;
import com.zhang.edu.realtime.bean.DimTableProcess;
import com.zhang.edu.realtime.utils.EnvUtil;
import com.zhang.edu.realtime.utils.KafkaUtil;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.streaming.api.datastream.BroadcastConnectedStream;
import org.apache.flink.streaming.api.datastream.BroadcastStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

public class DimSinkApp {
    public static void main(String[] args) throws Exception {
        // TODO 1. 环境准备及状态后端设置
        StreamExecutionEnvironment env = EnvUtil.getExecutionEnvironment(4);

        // TODO 2. 读取业务主流
        String topic = "topic_db";
        String groupId = "dim_sink_app";
        DataStreamSource<String> eduDS = env.fromSource(KafkaUtil.getKafkaConsumer(topic, groupId),
                WatermarkStrategy.noWatermarks(), "kafka_source");

        // TODO 3. 对主流数据进行ETL
        SingleOutputStreamOperator<JSONObject> jsonDS = eduDS.flatMap((FlatMapFunction<String, JSONObject>) (value, out) -> {
            try {
                JSONObject jsonObject = JSON.parseObject(value);
                String type = jsonObject.getString("type");
                if (!(type.equals("bootstrap-complete") || type.equals("bootstrap-start"))) {
                    // 需要的数据
                    out.collect(jsonObject);
                }
            } catch (Exception e) {
                e.printStackTrace();
                System.out.println("数据转换json错误");
            }
        }) ;
        jsonDS.print();

        // TODO 4 使用flinkCDC读取配置表数据
        // 4.1 FlinkCDC 读取配置表信息
        MySqlSource<String> mySqlSource = MySqlSource.<String>builder()
                .hostname("hadoop102")
                .port(3306)
                .databaseList("edu_config") // set captured database
                .tableList("edu_config.table_process") // set captured table
                .username("root")
                .password("000000")
                .deserializer(new JsonDebeziumDeserializationSchema()) // converts SourceRecord to JSON String
                .startupOptions(StartupOptions.initial())
                .build();

        // 4.2 封装为流
        DataStreamSource<String> mysqlDSSource = env.fromSource(mySqlSource, WatermarkStrategy.noWatermarks(), "MysqlSource");

        // TODO 5 将配置表数据创建为广播流
        MapStateDescriptor<String, DimTableProcess> tableConfigDescriptor = new MapStateDescriptor<>("table-process-state", String.class, DimTableProcess.class);
        BroadcastStream<String> broadcastDS = mysqlDSSource.broadcast(tableConfigDescriptor);

        // TODO 6. 连接流
        BroadcastConnectedStream<JSONObject, String> connectedStream = jsonDS.connect(broadcastDS);


        // TODO 7.对合并流分别处理
        SingleOutputStreamOperator<JSONObject> dimDS = connectedStream.process(new DimBroadcastFunction(tableConfigDescriptor));
        // TODO 8.调取维度数据写入到phoenix
        dimDS.addSink(new DimPhoenixSinkFunc());

        env.execute();
    }
}
